package PageRank;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class MapTest extends Mapper<Text, Text, Text, Text> {
    @Override
    protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
        //第一件事:要把上轮的投票关系.投票页面的pr值写出去
        //如果是首次运行程序,那么需要将每一页面的pr值设置为1
        int runCount = context.getConfiguration().getInt("runCount", 1);
        //投票页面
        String page = key.toString();
        //被投票页面
        String line = value.toString();
        //定义一个类去处理数据
        //这个类其实就是为每一行数据量身定做的
        Node node = null;
        //如果是第一次执行mr程序,需要将1.0赋值给每一个网页

        try {
            if (runCount == 1) {
                node = Node.getNode("1.0", line);
            } else {
                node = Node.getNode(line);
            }
            //这里写出的是页面关系
            context.write(key, new Text(node.toString()));
            //如果有出链
            //写出此链给他的出链的每一个pr值
            if (node.havingNode()) {
                double v = node.getPr() / node.getNearbyNodename().length;
                for (String out : node.getNearbyNodename()) {
                    context.write(new Text(out), new Text(String.valueOf(v)));
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
